package yuekao2.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import yuekao2.util.KafkaUtil;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class CustomizeCEP {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //1)	将所给附件中的数据生成jar包及配置文件上传至虚拟机,在kafka创建topic：ods_gmall_log，并运行jar包查看kafka中是否有数据（2分）
        //2)	创建maven项目，导入相关依赖，并创建目录com.bw.app,创建类BaseLogApp，以及flink流处理环境（2分）
        //3)	添加kafka作为数据源,过滤掉脏数据，并将页面日志、曝光日志、启动日志筛选出来存入到流中,新老访客状态标记修复思路（2分）
        // TODO: 2024/8/27 直接转换为JSON格式书写
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("ods_gmall_log"));
        SingleOutputStreamOperator<JSONObject> process = streamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                } catch (Exception e) {

                }
            }
        });
        SingleOutputStreamOperator<JSONObject> process1 = process.keyBy(x -> x.getJSONObject("common").getString("mid")).process(new ProcessFunction<JSONObject, JSONObject>() {
            ValueState<String> state;

            @Override
            public void open(Configuration parameters) {
                state = getRuntimeContext().getState(new ValueStateDescriptor<>("state", Types.STRING()));
            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String is_new = jsonObject.getJSONObject("common").getString("is_new");
                if ("1".equals(is_new) && state.value() != null) {
                    jsonObject.getJSONObject("common").put("is_new", 0);
                }
                state.update("老用户");
                collector.collect(jsonObject);
            }
        });
        //4)	将上述三条流分别sink到kafka 主题：dwd_start_log、dwd_display_log、dwd_page_log,创建类UserJumpApp类以及flink流处理环境（2分）
        OutputTag<String> start = new OutputTag<String>("start") {
        };
        OutputTag<String> display = new OutputTag<String>("display") {
        };
        OutputTag<String> page = new OutputTag<String>("page") {
        };

        SingleOutputStreamOperator<String> process2 = process1.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                if (jsonObject.containsKey("start")) {
                    context.output(start, jsonObject.toJSONString());
                }
                if (jsonObject.containsKey("display")) {
                    context.output(display, jsonObject.toJSONString());
                }
                if (jsonObject.containsKey("page")) {
                    context.output(page, jsonObject.toJSONString());
                }
            }
        });
        process2.getSideOutput(start).addSink(KafkaUtil.kafkaSink("dwd_start_log"));
        process2.getSideOutput(display).addSink(KafkaUtil.kafkaSink("dwd_display_log"));
        process2.getSideOutput(page).addSink(KafkaUtil.kafkaSink("dwd_page_log"));
        //5)	从kafka中读取dwd层的页面数据，过滤数据并转换类型为JSON对象,定义 CEP 匹配规则 （2分）
        DataStreamSource<String> pagelog = env.addSource(KafkaUtil.kafkaSource("dwd_page_log"));
        SingleOutputStreamOperator<JSONObject> pagets = pagelog.map(x -> JSON.parseObject(x))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> event.getLong("ts")));
        KeyedStream<JSONObject, String> keyby = pagets.keyBy(x -> x.getJSONObject("common").getString("mid"));
        //注意：定义 CEP 匹配规则
        //①规则一
        //跳出行为对应的页面日志必然为某一会话的首页，因此第一个规则判定 last_page_id 是否为 null，是则返回 true，否则返回 false；
        //②规则二
        //规则二和规则一之间的策略采用严格连续，要求二者之间不能有其它事件。判断 last_page_id 是否为 null，在数据完整有序的前提下，如果不是 null 说明本条日志的页面不是首页，可以断定它与规则一匹配到的事件同属于一个会
        // ，返回 false；如果是 null 则开启了一个新的会话，此时可以判定上一条页面日志所属会话为跳出会话，是我们需要的数据，返回 true；
        //③超时时间
        //超时时间内规则一被满足，未等到第二条数据则会被判定为超时数据。

        Pattern<JSONObject, JSONObject> within = Pattern.<JSONObject>begin("one").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject event) {
                return event.getJSONObject("common").getString("last_page_id") == null;
            }
        }).next("two").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject subEvent) {
                return subEvent.getJSONObject("common").getString("last_page_id") == null;
            }
        }).within(Time.minutes(10));

        PatternStream<JSONObject> patternStream = CEP.pattern(keyby, within);

        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("outputTag") {
        };

        SingleOutputStreamOperator<JSONObject> select = patternStream.select(outputTag, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                return map.get("one").get(0);
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                return map.get("one").get(0);
            }
        });
        DataStream<JSONObject> union = select.union(select.getSideOutput(outputTag));
//        union.print();
        //6)	把 Pattern 应用到流上，提取匹配上的事件以及超时事件,合并两个流并将数据写出到 Kafka（2分）
        union.map(x->JSON.toJSONString(x)).addSink(KafkaUtil.kafkaSink("dwd_union_log"));


        env.execute();
    }
}
